package com.huawei.spare.part.replace.rx;

import com.huawei.spare.part.replace.utils.Threads;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import java.io.File;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.X509Certificate;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javafx.beans.property.BooleanProperty;
import javafx.beans.property.SimpleBooleanProperty;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import lombok.AllArgsConstructor;
import lombok.Data;
import okhttp3.Call;
import okhttp3.Dispatcher;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okio.BufferedSink;
import okio.Okio;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public enum RxOkHttp {

  INSTANCE;

  private static final Logger log = LoggerFactory.getLogger(RxOkHttp.class);
  private static final long BUFFER_SIZE = 1024 * 1024;

  OkHttpClient client;
  ExecutorService threadPool;

  RxOkHttp() {
  }

  public void init() {
    if (client == null) {
      // create http request executor thread pool
      threadPool = Executors.newScheduledThreadPool(2, Threads.buildJobFactory("RxOkHttp-%d"));
      // Create an ssl socket factory with our all-trusting manager
      try {
        final TrustManager[] trustAllCerts = new TrustManager[] {
            new X509TrustManager() {
              @Override
              public void checkClientTrusted(X509Certificate[] x509Certificates, String s) {
              }

              @Override
              public void checkServerTrusted(X509Certificate[] x509Certificates, String s) {
              }

              @Override
              public X509Certificate[] getAcceptedIssuers() {
                return new X509Certificate[] {};
              }
            }
        };
        final SSLContext sslContext = SSLContext.getInstance("SSL");
        sslContext.init(null, trustAllCerts, new java.security.SecureRandom());
        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();

        this.client = new OkHttpClient().newBuilder()
            .dispatcher(new Dispatcher(threadPool))
            .retryOnConnectionFailure(true)
            .connectTimeout(30, TimeUnit.SECONDS)
            .readTimeout(120, TimeUnit.SECONDS)
            .sslSocketFactory(sslSocketFactory, (X509TrustManager) trustAllCerts[0])
            .hostnameVerifier((hostname, session) -> true).build();
      } catch (KeyManagementException | NoSuchAlgorithmException e) {
        log.error("Failed to initial rx client", e);
      }
    }
  }

  public static RxOkHttp get() {
    INSTANCE.init();
    return INSTANCE;
  }

  public Response syncGet(String url) throws IOException {
    final Request request = new Request.Builder().url(url).build();
    final Response response = client.newCall(request).execute();
    return response;
  }

  public Observable<Response> get(String url) {
    return Observable.defer((Callable<ObservableSource<Response>>) () -> {
      try {
        final Request request = new Request.Builder().url(url).build();
        final Response response = client.newCall(request).execute();
        if (response.isSuccessful()) {
          return Observable.just(response);
        }
        return Observable.error(new FailedResponseException(response));
      } catch (IOException e) {
        return Observable.error(e);
      }
    });
  }

  public Observable<Progress> downloadFile(final String downloadUrl, final File destFile) {
    return Observable.create((ObservableOnSubscribe<Progress>) emitter -> {
      final Call call = client.newCall(new Request.Builder().url(downloadUrl).build());
      final BooleanProperty finished = new SimpleBooleanProperty(false);
      emitter.setCancellable(() -> {
        if (!finished.get()) {
          log.info("Cancel download task for file: {}", destFile.getAbsolutePath());
          call.cancel(); // cancel download request
          destFile.delete();
        }
      });

      final Response response = call.execute();
      BufferedSink sink = Okio.buffer(Okio.sink(destFile));
      try {
        long contentLength = response.body().contentLength();
        long totalRead = 0;
        long lastRead;
        while ((lastRead = response.body().source().read(sink.buffer(), BUFFER_SIZE)) != -1) {
          totalRead += lastRead;
          sink.emitCompleteSegments();
          emitter.onNext(new Progress(totalRead, contentLength, false));
        }
        sink.writeAll(response.body().source());
        sink.close();
        response.body().close();

        finished.set(true);
        emitter.onNext(new Progress(totalRead, contentLength, true));
        emitter.onComplete();
      } catch (IOException e) {
        response.body().close();
        sink.close();
        if (!call.isCanceled()) {
          emitter.onError(e);
        }
      }
    });

    //        return Flowable.create((FlowableEmitter<Progress> emitter) -> {
    //            final File parent = new File(destFile.getParent());
    //            if(!parent.exists()) {
    //                parent.mkdirs();
    //            }
    //            final Call call = client.newCall(new Request.Builder().url(downloadUrl).build());
    //            final Response response = call.execute();
    //            BufferedSink sink = Okio.buffer(Okio.sink(destFile));
    //            try {
    //                long contentLength = response.body().contentLength();
    //                long totalRead = 0;
    //                long lastRead;
    //                while ((lastRead = response.body().source().read(sink.buffer(), BUFFER_SIZE)) != -1) {
    //                    totalRead += lastRead;
    //                    sink.emitCompleteSegments();
    //                    emitter.onNext(new Progress(totalRead, contentLength, false));
    //                }
    //                sink.writeAll(response.body().source());
    //                sink.close();
    //                response.body().close();
    //
    //                emitter.onNext(new Progress(totalRead, contentLength, true));
    //                emitter.onComplete();
    //            } catch (IOException e) {
    //                response.body().close();
    //                sink.close();
    //                emitter.onError(e);
    //            }
    //
    //            emitter.setCancellable(() -> call.cancel());
    //        }, BackpressureStrategy.LATEST).onBackpressureLatest();
  }

  @Data
  @AllArgsConstructor
  public class Progress {
    long totalRead;
    long contentLength;
    boolean complete;

    public double getProgress() {
      return totalRead * 1.0D / contentLength;
    }
  }

  public static class FailedResponseException extends Exception {

    private final Response response;

    public FailedResponseException(Response response) {
      super("Request Failed, http code: " + response.code());
      this.response = response;
    }

    public Response getResponse() {
      return response;
    }
  }

}
